package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.SimpleRedisLock;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.hmdp.constants.RedisConstants.LOCK_ORDER_KEY;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    private ISeckillVoucherService seckillVoucherService;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RedisIdWorker redisIdWorker;
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }
//    private BlockingQueue<VoucherOrder> orderQue = new ArrayBlockingQueue<>(1024*1024);
    private static final ExecutorService ORDER_POLL = Executors.newSingleThreadExecutor();
    private IVoucherOrderService proxy; //代理用于订单事务生效
    @PostConstruct
    private void init(){
        ORDER_POLL.submit(new VoucherClassHandler());
    }

    private class VoucherClassHandler implements Runnable{
        @Override
        public void run(){
            String que_name = "stream.orders";
            while(true){
                try{
                    // 1、从消息队列中取消息 xreadgroup group g1 c1 count 1 block 1000 streams stream.order >
                    List<MapRecord<String, Object, Object>> msgList = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(1)),
                            StreamOffset.create(que_name, ReadOffset.lastConsumed())
                    );
                    if(msgList==null || msgList.isEmpty()) continue;

                    // 2、保存消息
                    // 2.1 解析消息
                    MapRecord<String, Object, Object> msg = msgList.get(0); //string存的是消息id
                    Map<Object, Object> msgEntity = msg.getValue();
                    // 注意下面的map转voucherOrder，在存入消息时的字段一致性
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(msgEntity, new VoucherOrder(), true);
                    // 2.2 保存到数据库
                    handleVoucherOrder(voucherOrder);

                    // 3、确认消息ACK
                    stringRedisTemplate.opsForStream().acknowledge(que_name, "g1", msg.getId());
                }catch (Exception e){
                    while(true){
                        try{
                            // 1、从pending-list中取消息 xreadgroup group g1 c1 count 1 block 1000 streams stream.order 0
                            List<MapRecord<String, Object, Object>> msgList = stringRedisTemplate.opsForStream().read(
                                    Consumer.from("g1", "c1"),
                                    StreamReadOptions.empty().count(1),
                                    StreamOffset.create(que_name, ReadOffset.from("0"))
                            );
                            if(msgList==null || msgList.isEmpty()) break;

                            // 2、保存消息
                            // 2.1 解析消息
                            MapRecord<String, Object, Object> msg = msgList.get(0); //string存的是消息id
                            Map<Object, Object> msgEntity = msg.getValue();
                            // 注意下面的map转voucherOrder，在存入消息时的字段一致性
                            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(msgEntity, new VoucherOrder(), true);
                            // 2.2 保存到数据库
                            handleVoucherOrder(voucherOrder);
                            // 3、确认消息ACK
                            stringRedisTemplate.opsForStream().acknowledge(que_name, "g1", msg.getId());
                        }catch (Exception er2){
                            log.error("接收pending-list消息失败！");
                        }
                    }
                }
            }
        }
    }

    private void handleVoucherOrder(VoucherOrder task) {
        long userId = task.getUserId();
        // 1、拿到锁
        SimpleRedisLock rlock = new SimpleRedisLock(stringRedisTemplate, LOCK_ORDER_KEY+userId);
        boolean result = rlock.tryLock(100);
        if(!result){
            log.error("加锁失败！");
            return;
        }
        // 2、保存订单
        try{
           proxy.createVoucherOrder(task);
        }finally {
            rlock.unlock();
        }
    }


    @Override
    public Result seckillVoucher(Long voucherId) {
        long userId = UserHolder.getUser().getId();
        long orderId = redisIdWorker.nextId("order");
        // 1、由redis完成秒杀校验和下单, 并完成消息队列的保存
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(),
                voucherId.toString(), String.valueOf(userId), String.valueOf(orderId));
        // 校验redis返回的结果
        int res = result.intValue();
        if(res != 0 ){
            return Result.fail( res==1 ? "库存不足！": "重复下单！");
        }

        // 必须提前获取代理？
        IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
        this.proxy = proxy;

        return Result.ok(orderId);
    }

    @Transactional
    @Override
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        long userId = voucherOrder.getUserId();
        long voucherId = voucherOrder.getVoucherId();
        // 1、 查询是否下单
        Integer count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        if(count>0){
            log.error("用户已下单");
            return;
        }
        // 2、扣减库存（CAS避免超卖）
        boolean succ = seckillVoucherService.update()
                .setSql("stock = stock-1")
                .eq("voucher_id", voucherId).gt("stock", 0)
                .update();
        if(!succ){
            log.error("库存不足");
            return;
        }
        // 3、保存订单到数据库(订单id、用户id、优惠券id）
        save(voucherOrder);
    }
}
